package org.huangrui.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

/**
 * @Author hr
 * @Create 2024-10-22 0:51 
 */
object SparkStreaming06_State_Window {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val data = ssc.socketTextStream("localhost", 9999)
    val wordCount = data.flatMap(_.split(" ")).map(word => (word, 1))
    // TODO 窗口 : 其实就是数据的范围（时间）
    //      window方法可以改变窗口的数据范围（默认数据范围为采集周期）
    //      window方法可以传递2个参数
    //             第一个参数表示窗口的数据范围（时间）
    //             第二个参数表示窗口的移动幅度（时间），可以不用传递，默认使用的就是采集周期
    //      SparkStreaming是在窗口移动的时候计算的。
    val windowDs = wordCount.window(Seconds(3), Seconds(6))

    windowDs.reduceByKey(_ + _).print()
    ssc.start()
    ssc.awaitTermination()
  }
}
